/*
 * Copyright 2009 BioTeam Inc
 * 
 * Licensed under License GNU LESSER GENERAL PUBLIC LICENSE version 3.0 (the "License"); you may not
 * use this file except in compliance with the License. You may obtain a copy of
 * the License at
 * 
 * http://www.gnu.org/copyleft/lesser.html
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */
package net.bioteam.appweb.submit;

import java.io.File;
import java.io.InputStream;
import java.util.Date;
import java.util.concurrent.ConcurrentHashMap;

import javax.persistence.EntityManager;
import javax.persistence.EntityTransaction;

import net.bioteam.appweb.Config;
import net.bioteam.cloud.cluster.ActiveCluster;
import net.bioteam.cloud.cluster.ActiveHost;
import net.bioteam.cloud.cluster.ClusterStatus;
import net.bioteam.common.user.User;
import net.bioteam.common.utils.MyEntityManager;
import net.bioteam.common.utils.Utils;
import net.bioteam.appweb.ws.WsClient;

import org.apache.axis2.AxisFault;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

public class SoapStatusChecker implements RunStatusChecker
{
	public static final Log logger = LogFactory.getLog(SoapStatusChecker.class);
	
	public static final ConcurrentHashMap<Long, Boolean> downloadThreads=new ConcurrentHashMap<Long, Boolean>();

	// leave the transaction to JobCollector
	public void checkStatus(UnfinishedJob job)
	{
		EntityManager em = (EntityManager) MyEntityManager.getEntityManager();
		long clusterid = job.getClusterid();
		ActiveCluster ac = em.find(ActiveCluster.class, clusterid);

		// cluster is killed.
		if (ac == null)
		{
			logger.info("cluster for job " + job.getId() + " does not exist");
			job.setCode(RunStatusCode.ABORTED);
			job.setNote("cluster does not exist anymore");
			job.setFinishedTime(new Date());
			return;
		}

		if (job.getTrials() >= Config.JOB_TRIAL_NUM * 2)
		{
			// this is the third time
			logger.info("abort job " + job.getId() + " since we cloud not get job status in certain times");
			job.setCode(RunStatusCode.ABORTED);
			job.setNote("can not get job status in certain trials");
			job.setFinishedTime(new Date());
			return;
		}

		User user = em.find(User.class, Long.parseLong(job.getUsername()));
		ActiveHost head = ac.getHead();
		String serviceURL = "http://" + head.getExternalName() + ":8080/appweb/services/programws";
		WsClient client;

		try
		{
			client = new WsClient(user, serviceURL, null);
			String[] ret = client.getJobStatus(job.getRemoteJobId());
			RunStatusCode code = RunStatusCode.valueOf(ret[0]);
			if (code == RunStatusCode.FINISHED)
			{
				job.setCode(RunStatusCode.RETRIEVING);
				job.setFinishedTime(new Date());
				if (job.getJobid() == null && ret[1] != null)
					job.setJobid(ret[1]);
				job.setExitValue(Integer.parseInt(ret[2]));
				if (ret[3] != null)
					job.setNote(ret[3]);
				else
					job.setNote("finished. Plan to retrieve job result");
				ac.setStatus(ClusterStatus.RETRIEVING);
			} else
			{
				if (code != job.getCode())
					job.setCode(code);
				if (job.getJobid() == null && ret[1] != null)
					job.setJobid(ret[1]);
				// if (ret[3]!=null)
				// job.setNote(ret[3]);
			}
			return;
		} catch (AxisFault e)
		{
			logger.info(e.getMessage());
			job.setNote("Get status failed "+e.getMessage());
			job.setTrials(job.getTrials() + 1);
		}

	}

	public void postJobProcess(UnfinishedJob job)
	{

		if (job.getCode() != RunStatusCode.RETRIEVING)
			return;
		
		Boolean previous=downloadThreads.putIfAbsent(job.getId(), new Boolean(true));
		//another threads is downloading right now.
		if (previous!=null)
			return;
		
		new DownloadThread(job.getId()).start();
	}

}

class DownloadThread extends Thread
{
	UnfinishedJob job;
	long jobid;

	public DownloadThread(long jobid)
	{
		this.jobid = jobid;

	}

	/**
	 * need to manage transaction by ourselves since this is a separate thread
	 * 
	 */
	public void run()
	{
		EntityManager em = (EntityManager) MyEntityManager.getEntityManager();
		EntityTransaction tx = em.getTransaction();
		try
		{
			// job is from another EntityManager, it is unmanaged for this
			// entityManager.
			// so we need to reload it.
			job = em.find(UnfinishedJob.class, jobid);
			if (job.getCode()!=RunStatusCode.RETRIEVING)
				return;
			long clusterid = job.getClusterid();
			if (job.getTrials() >= Config.JOB_TRIAL_NUM * 3)
			{
				SoapStatusChecker.logger.info("abort job " + job.getId() + " since failed downloading in certain times");
				job.setCode(RunStatusCode.ABORTED);
				job.setNote("failed downloading in certain times");
				return;
			}

			ActiveCluster ac = em.find(ActiveCluster.class, clusterid);
			if (ac == null)
			{
				SoapStatusChecker.logger.info("cluster for job " + job.getId() + " does not exist");
				job.setCode(RunStatusCode.ABORTED);
				job.setNote("cluster does not exist anymore");
				return;
			}
			File workdir = new File(job.getWorkdir());
			User user = em.find(User.class, Long.parseLong(job.getUsername()));
			ActiveHost head = ac.getHead();
			String serviceURL = "http://" + head.getExternalName() + ":8080/appweb/services/programws";
			WsClient client;

			try
			{
				client = new WsClient(user, serviceURL, null);
				InputStream in = client.getJobFileAsZip(job.getRemoteJobId());
				Utils.unzipDir(in, workdir, false);

				job.setCode(RunStatusCode.FINISHED);
				job.setNote("result is retrieved successfully");
				// Let job collector to set the code when collect job
				// ac.setStatus(ClusterStatus.IDLE);
			} catch (Exception e)
			{

				SoapStatusChecker.logger.info(e.getMessage());
				job.setNote("downloading failed "+e.getMessage());
				job.setTrials(job.getTrials() + 1);
			}

		} finally
		{
			
			tx.begin();
			em.flush();
			tx.commit();
			SoapStatusChecker.downloadThreads.remove(jobid);
			MyEntityManager.close();
		}
	}

}
